{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Execution Plans, Lazy Evaluation and Caching"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### **Task:** calculate the sum of squares :  $$\\sum_{i=1}^n x_i^2 $$  \n",
    "The standard (or **busy**) way to do this is\n",
    "1. Calculate the square of each element.\n",
    "2. Sum the squares.\n",
    "\n",
    "This requires **storing** all intermediate results."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "<img alt=\"\" src=\"Figures/LazyEvaluation/Slide1.png\" style=\"height:455px;width:900px\" />"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "<p><img alt=\"\" src=\"Figures/LazyEvaluation/Slide2.png\" style=\"height:455px; width:900px\" /></p>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "<p><img alt=\"\" src=\"Figures/LazyEvaluation/Slide3.png\" style=\"height:455px; width:900px\" /></p>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### **lazy** evaluation:\n",
    " \n",
    "* **postpone** computing the square until result is needed.\n",
    "* No need to store intermediate results.\n",
    "* Scan through the data once, rather than twice."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "<p><img alt=\"\" src=\"Figures/LazyEvaluation/Slide4.png\" style=\"height:455px; width:900px\" /></p>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "<p><img alt=\"\" src=\"Figures/LazyEvaluation/Slide5.png\" style=\"height:455px; width:900px\" /></p>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "### Lazy Evaluation\n",
    "\n",
    "Unlike a regular python program, map/reduce commands do not always perform any computation when they are executed. Instead, they construct something called an **execution plan**. Only when a result is needed does the computation start. This approach is also called **lazy execution**.\n",
    "\n",
    "The benefit from lazy execution is in minimizing the the number of memory accesses. Consider for example the following map/reduce commands:\n",
    "```python\n",
    "A=RDD.map(lambda x:x*x).filter(lambda x: x%2==0)\n",
    "A.reduce(lambda x,y:x+y) \n",
    "```\n",
    "\n",
    "The commands defines the following plan. For each number `x` in the RDD:\n",
    "1. Compute the square of `x`\n",
    "2. Filter out `x*x` whose value is odd.\n",
    "3. Sum the elements that were not filtered out.\n",
    "\n",
    "A naive execution plan is to square all items in the RDD, store the results in a new RDD, then perform a filtering pass, generating a second RDD,  and finally perform the summation. Doing this will require iterating through the RDD three times, and creating 2 interim RDDs. As memory access is the bottleneck in this type of computation, the execution plan is slow.\n",
    "\n",
    "A better execution plan is to perform all three operations on each element of the RDD in sequence, and then move to the next element. This plan is faster because we iterate through the elements of the RDD only once, and because we don't need to save the intermediate results. We need to maintain only one variable: the partial sum, and as that is a single variable, we can use a CPU register.\n",
    "\n",
    "For more on RDDs and lazy evaluation see [here in the spark manual](http://spark.apache.org/docs/latest/programming-guide.html#rdd-operations)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Experimenting with Lazy Evaluation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "#### The `%%time` magic\n",
    "\n",
    "The `%%time` command is a *cell magic* which measures the execution time of the cell. We will mostly be interested in the wall time, which includes the time it takes to move data in the memory hierarchy.\n",
    "\n",
    "For more on jupyter magics [See here](https://ipython.org/ipython-doc/3/interactive/magics.html)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "### Preparations\n",
    "In the following cells we create an RDD and define a function which wastes some time and then returns `cos(i)`. We want the function to waste some time so that the time it takes to compute the `map` operation is significant."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 1,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [],
   "source": [
    "from pyspark import SparkContext\n",
    "sc = SparkContext(master=\"local[4]\")  #note that we set the number of workers to 3"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "We create an RDD with one million elements to amplify the effects of lazy evaluation and caching."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 2,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 0 ns, sys: 0 ns, total: 0 ns\n",
      "Wall time: 604 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "RDD=sc.parallelize(range(1000000))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "It takes about 01.-0.5 sec.  to create the RDD. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 3,
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(4) PythonRDD[1] at RDD at PythonRDD.scala:48 []\n",
      " |  ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:175 []\n"
     ]
    }
   ],
   "source": [
    "print(RDD.toDebugString().decode())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Define a computation\n",
    "The role of the function `taketime` is to consume CPU cycles."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 4,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [],
   "source": [
    "from math import cos\n",
    "def taketime(i):\n",
    "    [cos(j) for j in range(100)]\n",
    "    return cos(i)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 5,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 0 ns, sys: 0 ns, total: 0 ns\n",
      "Wall time: 52 µs\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "0.5403023058681398"
      ]
     },
     "execution_count": 5,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "%%time\n",
    "taketime(1)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Time units\n",
    "* 1 second = 1000 Milli-second ($ms$)\n",
    "* 1 Millisecond = 1000 Micro-second ($\\mu s$)\n",
    "* 1 Microsecond = 1000 Nano-second ($ns$)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Clock Rate\n",
    "One cycle of a 3GHz cpu takes $\\frac{1}{3} ns$\n",
    "\n",
    "`taketime(1000)` takes about 25 $\\mu s$ = 75,000 clock cycles."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### The `map` operation. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 6,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 0 ns, sys: 0 ns, total: 0 ns\n",
      "Wall time: 33.4 µs\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "Interm=RDD.map(lambda x: taketime(x))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "### How come so fast?\n",
    "* We expect this map operation to take 1,000,000 * 25 $\\mu s$ = 25 Seconds."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "* **Why** did the previous cell take just 29 $\\mu s$?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "* Because **no computation was done** "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "* The cell defined an **execution plan**, but did not execute it yet."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "**Lazy Execution** refers to this type of behaviour. The system delays actual computation until the latest possible moment. Instead of computing the content of the RDD, it adds the RDD to the **execution plan**.\n",
    "\n",
    "Using Lazy evaluation of a plan has two main advantages relative to immediate execution of each step:\n",
    "1. A single pass over the data, rather than multiple passes.\n",
    "2. Smaller memory footprint becase no intermediate results are saved."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Execution Plans\n",
    "At this point the variable `Interm` does not point to an actual data structure. Instead, it points to an execution plan expressed as a **dependence graph**. The dependence graph defines how the RDDs are computed from each other."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "The dependence graph associated with an RDD can be printed out using the method `toDebugString()`."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 7,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(4) PythonRDD[2] at RDD at PythonRDD.scala:48 []\n",
      " |  ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:175 []\n"
     ]
    }
   ],
   "source": [
    "print(Interm.toDebugString().decode())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "**Interm=** `(4) PythonRDD[2] at RDD at PythonRDD.scala:48 []`  \n",
    "`______(4)` corresponds to the number of partitions \n",
    "  \n",
    "**RDD   ** `=   |  ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:489 []`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "At this point only the two left blocks of the plan have been declared.\n",
    "<p><img alt=\"\" src=\"Figures/ExecutionPlan/Slide1.jpg\" style=\"height:120px; width:900px\" /></p>\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Actual execution\n",
    "The `reduce` command needs to output an actual output, **spark** therefor has to actually execute the `map` and the `reduce`. Some real computation needs to be done, which takes about 1 - 3 seconds (Wall time) depending on the machine used and on it's load."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 8,
   "metadata": {
    "scrolled": false,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "out= -0.2887054679684464\n",
      "CPU times: user 10 ms, sys: 10 ms, total: 20 ms\n",
      "Wall time: 25 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "print('out=',Interm.reduce(lambda x,y:x+y))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### How come so fast? (take 2)\n",
    "* We expect this map operation to take 1,000,000 * 25 $\\mu s$ = 25 Seconds.\n",
    "* Map+reduce takes only ~4 second. \n",
    "* Why?"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "* Because we have 4 workers, rather than one.\n",
    "* Because the measurement of a single iteration of `taketime` is an overestimate."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Executing a different calculation based on the same plan.\n",
    "The plan defined by `Interm` might need to be executed more than once.\n",
    "\n",
    "**Example:** compute the number of map outputs that are larger than zero."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 9,
   "metadata": {
    "scrolled": false,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "out= 500000\n",
      "CPU times: user 20 ms, sys: 0 ns, total: 20 ms\n",
      "Wall time: 22.7 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "print('out=',Interm.filter(lambda x:x>0).count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### The price of not materializing\n",
    "* The run-time (3.4 sec) is similar to that of the reduce (4.4 sec).\n",
    "* Because the intermediate results in `Interm` have not been saved in memory (materialized)\n",
    "* They need to be recomputed."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "The middle block: `Map(Taketime)` is executed twice. Once for each final step.\n",
    "<p><img alt=\"\" src=\"Figures/ExecutionPlan/Slide2.jpg\" style=\"height:200px; width:900px\" /></p>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Caching intermediate results\n",
    "* We sometimes want to keep the intermediate results in memory so that we can reuse them later without recalculating. * This will reduce the running time, at the cost of requiring more memory.\n",
    "* The method `cache()` indicates that the RDD generates in this plan should be stored in memory. Note that this is a **plan to cache**. The actual caching will be done only when the final result is needed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 10,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "CPU times: user 10 ms, sys: 0 ns, total: 10 ms\n",
      "Wall time: 47.1 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "Interm=RDD.map(lambda x: taketime(x)).cache()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "By adding the Cache after `Map(Taketime)`, we save the results of the map for the second computation.\n",
    "<p><img alt=\"\" src=\"Figures/ExecutionPlan/Slide3.jpg\" style=\"height:200px; width:900px\" /></p>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Plan to cache\n",
    "The definition of `Interm` is almost the same as before. However, the *plan* corresponding to `Interm` is more elaborate and contains information about how the intermediate results <span class=\"girk\">will be</span> cached and replicated.\n",
    "\n",
    "Note that `PythonRDD[4]` is now <span class=\"mark\">[Memory Serialized 1x Replicated]</span>"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 11,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "(4) PythonRDD[5] at RDD at PythonRDD.scala:48 [Memory Serialized 1x Replicated]\n",
      " |  ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:489 [Memory Serialized 1x Replicated]\n"
     ]
    }
   ],
   "source": [
    "print(Interm.toDebugString().decode())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "skip"
    }
   },
   "source": [
    "#### Comparing plans with and without cache\n",
    "Plan with Cache\n",
    "```\n",
    "(4) PythonRDD[33] at RDD at PythonRDD.scala:48 [Memory Serialized 1x Replicated]\n",
    " |  ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:489 [Memory Serialized 1x Replicated]\n",
    "```  \n",
    "Plan without Cache\n",
    "```\n",
    "(4) PythonRDD[2] at RDD at PythonRDD.scala:48 []\n",
    " |  ParallelCollectionRDD[0] at parallelize at PythonRDD.scala:489 []\n",
    "```\n",
    "The difference is that the plan for both RDDs includes **[Memory Serialized 1x Replicated]** which is the plan to materialize both RDDs when they are computed."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Creating the cache\n",
    "The following command executes the first map-reduce command **and** caches the result of the `map` command in memory. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 12,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "out= -0.2887054679684655\n",
      "CPU times: user 5.43 ms, sys: 3.79 ms, total: 9.22 ms\n",
      "Wall time: 3.59 s\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "print('out=',Interm.reduce(lambda x,y:x+y))"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Using the cache\n",
    "This time `Interm` is cached. Therefor the second use of `Interm` is much faster than when we did not use `cache`: 0.25 second instead of 1.9 second. (your milage may vary depending on the computer you are running this on)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 13,
   "metadata": {
    "scrolled": true,
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "out= 500000\n",
      "CPU times: user 5.31 ms, sys: 2.97 ms, total: 8.28 ms\n",
      "Wall time: 121 ms\n"
     ]
    }
   ],
   "source": [
    "%%time\n",
    "print('out=',Interm.filter(lambda x:x>0).count())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Summary\n",
    "* Spark uses **Lazy Evaluation** to save time and space.\n",
    "* When the same RDD is needed as input for several computations, it can be better to keep it in memory, also called `cache()`.\n",
    "* Next Video, Partitioning and Gloming"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Partitioning and Gloming"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "* When an RDD is created, you can specify the number of partitions.\n",
    "* The default is the number of workers defined when you set up `SparkContext`"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 14,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "4\n"
     ]
    }
   ],
   "source": [
    "A=sc.parallelize(range(1000000))\n",
    "print(A.getNumPartitions())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "We can repartition `A` into a different number of partitions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 15,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10\n"
     ]
    }
   ],
   "source": [
    "D=A.repartition(10)\n",
    "print(D.getNumPartitions())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "We can also define the number of partitions when creating the RDD."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 16,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "10\n"
     ]
    }
   ],
   "source": [
    "A=sc.parallelize(range(1000000),numSlices=10)\n",
    "print(A.getNumPartitions())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### Why is the #Partitions important?\n",
    "* They define the unit the executor works on.\n",
    "* You should have at least as pany partitions as workers.\n",
    "* Smaller partitions can allow more parallelization."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Repartitioning for Load Balancing"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Suppose we start with 10 partitions, all with exactly the same number of elements"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 17,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000, 100000]\n"
     ]
    }
   ],
   "source": [
    "A=sc.parallelize(range(1000000))\\\n",
    "    .map(lambda x:(x,x)).partitionBy(10)\n",
    "print(A.glom().map(len).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "* Suppose we want to use `filter()` to select some of the elements in `A`.\n",
    "* Some partitions might have more elements remaining than others."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 18,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[100000, 0, 0, 0, 0, 100000, 0, 0, 0, 0]\n"
     ]
    }
   ],
   "source": [
    "#select 10% of the entries\n",
    "B=A.filter(lambda pair: pair[0]%5==0)\n",
    "# get no. of partitions\n",
    "print(B.glom().map(len).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "* Future operations on `B` will use only two workers.\n",
    "* The other workers will do nothing,   \n",
    "  because their partitions are empty."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "* To fix the situation we need to repartition the RDD.  \n",
    "* One way to do that is to repartition using a new key."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "source": [
    "* The method **`.partitionBy(k)`** expects to get a **`(key,value)`** RDD where keys are integers.  \n",
    "* Partitions the RDD into **`k`** partitions.\n",
    "* The element **`(key,value)`** is placed into partition no.  **`key % k`**"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 19,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000]\n"
     ]
    }
   ],
   "source": [
    "C=B.map(lambda pair:(pair[1]/10,pair[1])).partitionBy(10) \n",
    "print(C.glom().map(len).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "Another approach is to use random partitioning using **`repartition(k)`**\n",
    "* An **advantage** of random partitioning is that it does not require defining a key.\n",
    "* A **disadvantage** of random partitioning is that you have no control on the partitioning."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 20,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000]\n"
     ]
    }
   ],
   "source": [
    "C=B.repartition(10)\n",
    "print(C.glom().map(len).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "### Glom()\n",
    "* In general, spark does not allow the worker to refer to specific elements of the RDD.\n",
    "* Keeps the language clean, but can be a major limitation."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "* **glom()** transforms each partition into a tuple (immutabe list) of elements.\n",
    "* Creates an RDD of tules. One tuple per partition.\n",
    "* workers can refer to elements of the partition by index. \n",
    "* but you cannot assign values to the elements, the RDD is still immutable."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "* Now we can understand the command used above to count the number of elements in each partition.\n",
    "* We use `glom()` to make each partition into a tuple.\n",
    "* We use `len` on each partition to get the length of the tuple - size of the partition.\n",
    "* We `collect` the results to print them out."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 21,
   "metadata": {
    "slideshow": {
     "slide_type": "fragment"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000, 20000]\n"
     ]
    }
   ],
   "source": [
    "print(C.glom().map(len).collect())"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "source": [
    "### A more elaborate example\n",
    "There are many things that you can do using `glom()`.  \n",
    "\n",
    "Below is an example, can you figure out what it does?"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": 22,
   "metadata": {
    "slideshow": {
     "slide_type": "subslide"
    }
   },
   "outputs": [
    {
     "name": "stdout",
     "output_type": "stream",
     "text": [
      "[(0, 100000, 999990), None, None, None, None, (5, 100000, 999990), None, None, None, None]\n"
     ]
    }
   ],
   "source": [
    "def getPartitionInfo(G):\n",
    "    d=0\n",
    "    if len(G)>1: \n",
    "        for i in range(len(G)-1):\n",
    "            d+=abs(G[i+1][1]-G[i][1]) # access the glomed RDD that is now a  list\n",
    "        return (G[0][0],len(G),d)\n",
    "    else:\n",
    "        return(None)\n",
    "\n",
    "output=B.glom().map(lambda B: getPartitionInfo(B)).collect()\n",
    "print(output)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "collapsed": true,
    "slideshow": {
     "slide_type": "slide"
    }
   },
   "source": [
    "## Summary\n",
    "* We learned why partitions are important and how to control them.\n",
    "* We Learned how `glom()` can be used to allow workers to access their partitions as lists."
   ]
  }
 ],
 "metadata": {
  "anaconda-cloud": {},
  "celltoolbar": "Slideshow",
  "hide_input": false,
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.6.3"
  },
  "toc": {
   "base_numbering": 1,
   "nav_menu": {
    "height": "306px",
    "width": "252px"
   },
   "number_sections": true,
   "sideBar": true,
   "skip_h1_title": false,
   "title_cell": "Table of Contents",
   "title_sidebar": "Contents",
   "toc_cell": false,
   "toc_position": {
    "height": "calc(100% - 180px)",
    "left": "10px",
    "top": "150px",
    "width": "251px"
   },
   "toc_section_display": "block",
   "toc_window_display": true
  },
  "varInspector": {
   "cols": {
    "lenName": 16,
    "lenType": 16,
    "lenVar": 40
   },
   "kernels_config": {
    "python": {
     "delete_cmd_postfix": "",
     "delete_cmd_prefix": "del ",
     "library": "var_list.py",
     "varRefreshCmd": "print(var_dic_list())"
    },
    "r": {
     "delete_cmd_postfix": ") ",
     "delete_cmd_prefix": "rm(",
     "library": "var_list.r",
     "varRefreshCmd": "cat(var_dic_list()) "
    }
   },
   "types_to_exclude": [
    "module",
    "function",
    "builtin_function_or_method",
    "instance",
    "_Feature"
   ],
   "window_display": false
  }
 },
 "nbformat": 4,
 "nbformat_minor": 1
}
